/*******************************************************************************
 * Copyright (c) 2008 Sylvain Sicard.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser Public License v2.1
 * which accompanies this distribution, and is available at
 * http://www.gnu.org/licenses/old-licenses/gpl-2.0.html
 * 
 * Contributors:
 *     Sylvain Sicard - initial API and implementation
 ******************************************************************************/
package fr.jade.fraclite.orb.protocols;

import java.lang.reflect.Method;

import org.jgroups.Channel;
import org.jgroups.JChannelFactory;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.util.Promise;

import fr.jade.fraclite.orb.api.Marshaller;
import fr.jade.fraclite.orb.io.BasicMarshaller;
import fr.jade.fraclite.orb.io.messages.AdminMessage;
import fr.jade.fraclite.orb.io.messages.MethodCall;
import fr.jade.fraclite.orb.io.messages.MethodReply;
import fr.jade.fraclite.orb.io.messages.MulticastRpcState;

/**
 * @author Sylvain Sicard
 */
public class MulticastRpcClient extends ReceiverAdapter {
  protected static JChannelFactory factoryStub     = new JChannelFactory();
  protected static String          nullCallId      = "";
  protected Channel                channel;
  protected Marshaller             rmiMarshaller;
  protected Marshaller             rawMarshaller;
  protected Object                 result;
  protected String                 currentCallId   = nullCallId;
  protected boolean                resultOk        = false;
  protected Boolean                init            = false;
  protected String                 groupId;
  protected Long callCounter = 0L;
  protected Long stubId;
  Promise<Boolean> state_promise=new Promise<Boolean>();
  
  int responseCounter = 0;

  public MulticastRpcClient(Long stubId, String groupId, Marshaller marshaller) {
    this.rmiMarshaller = marshaller;
    this.rawMarshaller = new BasicMarshaller();
    this.groupId = groupId;
    this.stubId = stubId;
  }

  protected void init() {
    init = true;
    state_promise.reset();
    try {
      factoryStub.setMultiplexerConfig("ha-stacks.xml");
      //factoryStub.setMultiplexerConfig("udp.xml");
      // channel = factoryStub.createChannel("udp");
      //channel = factoryStub.createMultiplexerChannel("udp", groupId);
      channel = factoryStub.createMultiplexerChannel("udp", groupId);
      channel.setOpt(Channel.LOCAL, false);
      channel.setReceiver(this);
      channel.connect(groupId);
      //initState();
    } catch (Exception e1) {
      e1.printStackTrace();
    }
  }

  public Object callRemoteMethods(Method mth, Object[] args) throws Exception {
    if (!init) init();
    String callId = stubId+"@"+callCounter++;
    //System.out.println("->"+callId);
    MethodCall mc = new MethodCall(callId, mth, rmiMarshaller.objectToByteBuffer(args));
    byte[] buf = rawMarshaller.objectToByteBuffer(mc);
    //System.out.println("Call "+mc);
    Message msg = new Message(null, null, buf);
    channel.send(msg);
    resultOk = false;
    currentCallId = callId;
    responseCounter = 0;
    synchronized (currentCallId) {
      while (!resultOk)
        currentCallId.wait();
      currentCallId = nullCallId;
    }
    // System.out.println("Call "+callId+" return "+result);
    return result;
  }

  /* -------------------- Interface MessageListener ------------------- */

  @Override
  public void setState(byte[] state){
    System.out.println("[stub] setState");
    try {
      MulticastRpcState s = (MulticastRpcState)rawMarshaller.objectFromByteBuffer(state);
      callCounter = s.getCallCounter(stubId);
      //callCounter++;
      System.out.println("Stub state - "+stubId+":"+callCounter);
    } catch (Exception e) {
      e.printStackTrace();
    }
    state_promise.setResult(Boolean.TRUE);
  }
  
  public void initState(){
    try {
      Message msg = new Message(null, null, rawMarshaller.objectToByteBuffer(AdminMessage.GETSTATE));
      channel.send(msg);
    } catch (Exception e) {
      e.printStackTrace();
    }
    Boolean result=state_promise.getResult(10000);
    if(result == null) {
        System.out.println("[stub] setState() never got called");
    } else {
      System.out.println("[stub] setState() was called");
    }
    
  }
  
  @Override
  public void receive(Message msg) {
    try {
      // System.out.println("Receive a message");
      Object o = rawMarshaller.objectFromByteBuffer(msg.getBuffer());
      if (o instanceof MethodReply) {
        MethodReply reply = (MethodReply) o;
        // System.out.println("methodReply "+reply);
        synchronized (currentCallId) {
          if (currentCallId.equals(reply.getMethodCallId())) {
            if (responseCounter == 0) {
              responseCounter++;
            } else {
              // System.out.println("receive my reply");
              result = rmiMarshaller.objectFromByteBuffer(reply.getMethodResult());
              resultOk = true;
              currentCallId.notifyAll();
            }
          } else {
            // System.out.println("Receive a duplicate method reply or reply for methodCall from another stub");
            // ignore
            // - duplicate reply
            // - or reply for methodCall from another stub
          }
        }
      } if (o instanceof MulticastRpcState){
        setState(msg.getBuffer());
      }else {
        // System.out.println("Receive a message type unknown");
        // ignore
        // message type unknown
        //System.out.println("stub ignore: "+msg.printHeaders());
      }
    } catch (Exception e1) {
      e1.printStackTrace();
    }
  }
  /*--------------------------------------------------------------*/

}
